package com.hna.face.mqtt;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.hna.face.bussiness.DataMessageHandler;
import com.hna.face.bussiness.MessageHandler;
import com.hna.face.bussiness.OnlineMessageHandler;
import com.hna.face.bussiness.WillMessageHandler;
import com.hna.face.config.SysConfig;
import com.hna.face.service.IFaceRecognizeResultService;
import com.hna.face.service.IFaceVehicleService;

/**
 * 发布消息的回调类
 * 
 * 必须实现MqttCallback的接口并实现对应的相关接口方法 ◦CallBack 类将实现
 * MqttCallBack。每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。在回调中，将它用来标识已经启动了该回调的哪个实例。
 * ◦必须在回调类中实现三个方法：
 * 
 * public void messageArrived(MqttTopic topic, MqttMessage message) 接收已经预订的发布。
 * 
 * public void connectionLost(Throwable cause) 在断开连接时调用。
 * 
 * public void deliveryComplete(MqttDeliveryToken token)) 接收到已经发布的 QoS 1 或 QoS 2
 * 消息的传递令牌时调用。 ◦由 MqttClient.connect 激活此回调。
 * 
 */

@Service
public class PushCallback implements MqttCallback {
	private final static Logger log = LoggerFactory.getLogger(PushCallback.class);

	@Autowired
	private WillMessageHandler willMessageHandler;

	@Autowired
	private DataMessageHandler dataMessageHandler;

	@Autowired
	private OnlineMessageHandler onlineMessageHandler;

	private static final int POOL_SIZE = 100;
	private ExecutorService exe = Executors.newFixedThreadPool(POOL_SIZE);

	@Override
	public void connectionLost(Throwable cause) {
		// 连接丢失后，一般在这里面进行重连
		log.error("连接断开",cause);
	}

	@Override
	public void messageArrived(String topic, MqttMessage message) throws Exception {
		// TODO Auto-generated method stub
		// subscribe后得到的消息会执行到这里面
		log.info("接收消息Qos:" + message.getQos());
		log.info("接收消息内容:" + new String(message.getPayload()));
		try {
			MessageHandler handler = null;
			if (topic.equals(SysConfig.mqttWill)) {
				log.info("topic:" + SysConfig.mqttWill);
				willMessageHandler.setMessage(message);
				log.info("2");
				handler = willMessageHandler;
			} else if (topic.equals(SysConfig.mqttOnline)) {
				log.info("topic:" + SysConfig.mqttOnline);
				onlineMessageHandler.setMessage(message);
				handler = onlineMessageHandler;
			} else if (topic.equals(SysConfig.mqttData)) {
				log.info("data:" + SysConfig.mqttData);
				dataMessageHandler.setMessage(message);
				handler = dataMessageHandler;
			}
			if (handler == null) {
				log.error("handler is null");
				return;
			}
			exe.submit(handler);
		} catch (Exception e) {
			log.error("", e);
		}

	}

	@Override
	public void deliveryComplete(IMqttDeliveryToken token) {
		log.info("deliveryComplete---------" + token.isComplete());
	}

}
